package jupitermouse.site.example.java.batch;

import jupitermouse.site.example.scala.batch.DBUtils;
import org.apache.flink.api.common.functions.MapFunction;

import org.apache.flink.api.common.functions.MapPartitionFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

public class DataSetTransformationJavaApp {
    public static void main(String[] args) throws Exception {
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

//        mapFuncation(env);

//        filterFuncation(env);

        mapPartitionFunction(env);
    }

    public static void mapFuncation(ExecutionEnvironment env) throws Exception {
        DataSource<Integer> dataSource = env.fromCollection(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));
        dataSource.print();
//        dataSource.map(x -> x+1).print();
        dataSource.map(new MapFunction<Integer, Integer>() {
            @Override
            public Integer map(Integer value) throws Exception {
                return value + 1;
            }
        }).print();
    }

    public static void filterFuncation(ExecutionEnvironment env) throws Exception {
        DataSource<Long> dataSource = env.fromCollection(Arrays.asList(1L, 2L, 3L, 4L, 5L, 6L, 7L, 8L, 9L, 10L));
        dataSource.print();
        dataSource.map(x -> x+1).filter(x -> x>5).print();
    }

    public static void mapPartitionFunction(ExecutionEnvironment env) throws Exception {
        List<String> list = new ArrayList<>();
        for(int i=0;i<100;i++){
            list.add("student["+i+"]");
        }
        DataSource<String> data = env.fromCollection(list);


/*        dataSource.map(new MapFunction<Long, Long>() {
            @Override
            public Long map(Long value) throws Exception {
                String connection = DBUtils.getConnection();
                System.out.println("connection = ["+value+"]");

                DBUtils.returnConnection(connection);
                return null;
            }
        }).print();*/
        data.mapPartition(new MapPartitionFunction<String, String>() {
            @Override
            public void mapPartition(Iterable<String> values, Collector<String> out) throws Exception {
                String conn = DBUtils.getConnection();
                System.out.println("connection ["+conn+"]");
                DBUtils.returnConnection(conn);
            }
        }).print();
    }
}
